package kuoge.practice.kafka.stream.member.activity.conversionrate;

import com.alibaba.fastjson.JSON;
import io.netty.util.HashedWheelTimer;
import io.netty.util.TimerTask;
import kuoge.practice.kafka.stream.member.KafkaProducerFactory;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.time.Duration;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;

import static kuoge.practice.kafka.stream.member.TopicConstants.ITEM_PURCHASE_TOPIC;
import static kuoge.practice.kafka.stream.member.TopicConstants.ITEM_VISIT_TOPIC;

/**
 * @Description
 */
public class EventProducer {
    private static final Producer<String, String> kafkaProducer = KafkaProducerFactory.getInstance();
    private final HashedWheelTimer hashedWheelTimer = new HashedWheelTimer();

    void startSendEvent() {
        item_001_event();

        item_002_event();

        System.out.println("start send event success");
    }

    private void item_001_event() {
        TimerTask task = timeout -> {
            final UserVisitItemEvent userVisitItemEvent = new UserVisitItemEvent("user_a", "item_001");
            kafkaProducer.send(new ProducerRecord<>(ITEM_VISIT_TOPIC, userVisitItemEvent.getItemCode(), JSON.toJSONString(userVisitItemEvent)), (rs, ex) -> System.out.println(rs.offset() + ", " + ex));
            if (ThreadLocalRandom.current().nextInt(10) > 5) {
                Thread.sleep(Duration.ofSeconds(ThreadLocalRandom.current().nextInt(3, 10)).toMillis());
                final PurchaseItemEvent purchaseItemEvent = new PurchaseItemEvent("user_a", "item_001", "order_1");
                kafkaProducer.send(new ProducerRecord<>(ITEM_PURCHASE_TOPIC, purchaseItemEvent.getItemCode(), JSON.toJSONString(purchaseItemEvent)), (rs, ex) -> System.out.println(rs.offset() + ", " + ex));
            }
            System.out.println("send user_a visit item event, thread:" + Thread.currentThread().getName());
            item_001_event();
        };
        hashedWheelTimer.newTimeout(task, 1000, TimeUnit.MILLISECONDS);
    }

    private void item_002_event() {
        TimerTask task = timeout -> {
            final UserVisitItemEvent userVisitItemEvent = new UserVisitItemEvent("user_b", "item_002");
            kafkaProducer.send(new ProducerRecord<>(ITEM_VISIT_TOPIC, userVisitItemEvent.getItemCode(), JSON.toJSONString(userVisitItemEvent)), (rs, ex) -> System.out.println(rs.offset() + ", " + ex));
            if (ThreadLocalRandom.current().nextInt(10) > 7) {
                Thread.sleep(Duration.ofSeconds(ThreadLocalRandom.current().nextInt(3, 10)).toMillis());
                final PurchaseItemEvent purchaseItemEvent = new PurchaseItemEvent("user_b", "item_002", "order_2");
                kafkaProducer.send(new ProducerRecord<>(ITEM_PURCHASE_TOPIC, purchaseItemEvent.getItemCode(), JSON.toJSONString(purchaseItemEvent)), (rs, ex) -> System.out.println(rs.offset() + ", " + ex));
            }
            System.out.println("send user_b visit item event, thread:" + Thread.currentThread().getName());
            item_002_event();
        };
        hashedWheelTimer.newTimeout(task, 3000, TimeUnit.MILLISECONDS);
    }

    void stopSendEvent() {
        hashedWheelTimer.stop();
    }

}
